package org.arch_learn.dubbo_demo.dubbo_spi_router_demo;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class ReadyRestartInstances implements PathChildrenCacheListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(ReadyRestartInstances.class);

    private static final String LISTEN_PATH = "/dubbo_demo/restart/instances";

    private final CuratorFramework zkClient;

    //当节点变化时，给这个集合赋值 重启机器的信息列表
    private volatile Set<String> restartInstances = new HashSet<>();

    private ReadyRestartInstances(CuratorFramework client) {
        this.zkClient = client;
    }

    public static ReadyRestartInstances create() {
        CuratorFramework client = ZookeeperClients.client();
        //检查监听路径是否存在
        final Stat stat;
        try {
            stat = client.checkExists().forPath(LISTEN_PATH);
            if (stat == null) {
                client.create().creatingParentsIfNeeded().forPath(LISTEN_PATH);
            }
        } catch (Exception e) {
            LOGGER.error("确保基础路径存在！");
            e.printStackTrace();
        }
        ReadyRestartInstances readyRestartInstances = new ReadyRestartInstances(client);
        //创建一个nodeCache
        PathChildrenCache nodeCache = new PathChildrenCache(client, LISTEN_PATH, false);
        //给nodeCache添加监听器
        nodeCache.getListenable().addListener(readyRestartInstances);
        try {
            nodeCache.start();
        } catch (Exception e) {
            LOGGER.error("启动路径监听失败！");
            e.printStackTrace();
        }
        return readyRestartInstances;
    }


    /**
     * 返回应用名和主机拼接后的字符串
     *
     * @param appName
     * @param host
     * @return
     */
    private String buildApplicationAndInstanceString(String appName, String host) {
        return appName + "_" + host;
    }

    /**
     * 增加重启实例的配置信息方法
     *
     * @param appName
     * @param host
     * @throws Exception
     */
    public void addRestartingInstance(String appName, String host) throws Exception {
        zkClient.create()
                .creatingParentsIfNeeded()
                .forPath(LISTEN_PATH + "/" + buildApplicationAndInstanceString(appName, host));
    }

    /**
     * 删除重启实例的配置信息方法
     *
     * @param appName
     * @param host
     * @throws Exception
     */
    public void removeRestartingInstance(String appName, String host) throws Exception {
        zkClient.delete()
                .forPath(LISTEN_PATH + "/" + buildApplicationAndInstanceString(appName, host));
    }

    /**
     * 判断节点信息是否存在于restartInstances中
     *
     * @param appName
     * @param host
     * @return
     */
    public boolean hasRestartingInstance(String appName, String host) {
        return restartInstances.contains(buildApplicationAndInstanceString(appName, host));
    }

    @Override
    public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
        //查询出监听路径下所有的目录配置信息
        List<String> restartingInstancesStr = zkClient.getChildren().forPath(LISTEN_PATH);
        //给restartInstances赋值
        if (CollectionUtils.isEmpty(restartingInstancesStr)) {
            restartInstances = Collections.EMPTY_SET;
        } else {
            restartInstances = new HashSet<>(restartingInstancesStr);
        }
    }
}
